/**
 * AggActiveThread.java
 * ************************** 
 * @date Mar 31, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.logging.Level;
import java.util.logging.Logger;

import messages.Message;

import gossipServices.aggregation.functions.AggregationUpdateFunction;
import gossipServices.basic.nodeDescriptors.ConcreteNodeDescriptor;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.view.PartialView;
import gossipServices.gossipMessages.AggregationMessage;
import gossipServices.gossipThread.ActiveGossipThread;
import gossipServices.network.NetworkServicesAPI;
import gossipServices.pss.PeerSamplingServiceAPI;

/**
 * Description: 
 *
 */
public class AggActiveThreadOld extends ActiveGossipThread {

	protected PeerSamplingServiceAPI pss;
	protected NetworkServicesAPI nia;
	protected EpochHandler epochHandler;
	protected AggregationUpdateFunction updateFunc;
	protected static SingleValueContainer localValue;
	protected Double initialValue;
	
	private Logger log;
	
	/**
	 * @param name
	 * @param cycleLenghtMsec
	 * @param view
	 */
	public AggActiveThreadOld(String name, long cycleLenghtMsec, PartialView view, 
			PeerSamplingServiceAPI pss, NetworkServicesAPI nia, 
			EpochHandler epochHandler , AggregationUpdateFunction updateFunc) {
		super(name, cycleLenghtMsec, view);
		this.pss = pss;
		this.nia = nia;
		this.epochHandler = epochHandler;
		this.updateFunc = updateFunc;
		
		log = Logger.getLogger("ActiveAgg-log");
		log.setLevel(Level.OFF);
	}

	@Override
	public void activeThreadInstructions() throws InterruptedException {
		NodeDescriptor q;
		do{
			 q = pss.getPeer();
			 /* In the Aggregation a node can't extract itself!
			 (or we will lose values in the evaluations)
			 */
		}
		while(nia.getNodeName().equals(q.getNodeName()));
		NodeDescriptor me;
		try {
			me = new ConcreteNodeDescriptor(nia.getNodeName(), 
					InetAddress.getLocalHost(), new Integer(0));
			Message msg;
			synchronized (localValue) {
				msg = new AggregationMessage(me, q, localValue,
						AggActiveThreadOld.class,
						epochHandler.getEpoch(), 
						epochHandler.getRemainingCycles());				
			}
			nia.send(msg, q);
			log.finer("sending msg: "+((AggregationMessage)msg)+" to "+q);
			AggregationMessage msgQ = (AggregationMessage) nia.receive(q);
			
			/*
			 * WARNING: 
			 *  option 1) make the receive(q) a blocking call
			 *    |-> may lead to a deadlock if messages are sent to crashed nodes
			 *   
			 *  option 2) verify the incoming message leaving unchanged the current value
			 */
			if(msgQ != null){				
				log.finer("received: "+msgQ);
//				log.finer("exchanging");
//				Double sq = msgQ.getValue();					
//				activeExchange(sq);
				
				switch(epochHandler.checkEpochs(msgQ, epochHandler)){
				case exchange:
					System.out.println("Epoch: "+epochHandler.getEpoch()
							+" cycle: "+ epochHandler.getElapsedCycles()+
							"exchanging with "+msgQ.getSender()+"\n before exchange: "+localValue);
					SingleValueDouble sq = msgQ.getSingleValue();					
					activeExchange(sq.getValue());
					break;

				case sendCorrectEpoch:
					Message corrMsg = epochHandler.composeCorrectionMessage(me, msgQ.getSender(),
							AggPassiveThreadOld.class);
					log.finer("send correction MSG: "+ corrMsg +" to "+msgQ.getSender() );
					nia.send(corrMsg, msgQ.getSender());
					break;

				case updateCurrentEpoch:
					log.finer("updating current epoch from "+msgQ);
					epochHandler.updateCurrentEpoch(msgQ);
					break;
				
				default:

				}
			}
			
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}catch(AggregationRestartException e){
			e.printStackTrace();
			restart(e.getRemainingCycles());
		}
		
	}


	/**
	 * @param sq
	 */
	private void activeExchange(Double sq) {
		synchronized (localValue) {

			// **************
			// *** UPDATE ***
			localValue.setValue(updateFunc.update(localValue.getValue(), sq.doubleValue()));
			// **************
			
			System.out.println("Aggregation actual value: "+localValue+" epoch "+
					+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());
		}
		
		/*
		 * EPOCH Cycles Handling
		 * suppose that only the complete send + receive with 
		 * a successful exchange leads to an "correct" elapsed cycle
		 */
		try {
			epochHandler.elapsedCycle();
			log.finer(">>> EPOCH: "+epochHandler.getEpoch()+" cycle: "+epochHandler.getElapsedCycles());		
		} catch (AggregationRestartException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			restart(e.getRemainingCycles());
		}
	}

	/**
	 * @param remainingCycles
	 */
	private void restart(int waitCycles) {
		log.finer("restarting");
		synchronized (localValue) {
			localValue.setValue(initialValue);
			System.out.println("Restarting, actual value: "+localValue);
		}
		try {
			if(waitCycles > 0){
				log.finer("sleep "+waitCycles * cycleLenght_msec+" msec");
				// wait(waitCycles) || wake up at the first correct exchanging message
				Thread.sleep(waitCycles * cycleLenght_msec); //<-make it more accurate!!!!
				log.finer("wake up!!");
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public final Double getInitialValue() {
		return initialValue;
	}

	public final void setInitialValue(Double initialValue) {
		this.initialValue = initialValue;
	}


	public final SingleValueContainer getValue() {
		return localValue;
	}

	public final void setValue(SingleValueContainer inValue) {
		localValue = inValue;
	}

}
